package com.homedo.rxj;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;

import java.util.Random;

public class RxjTest1 {
    public static void main(String[] args) {

        createObserver().doOnComplete(new Action() {
            @Override
            public void run() throws Exception {
                System.out.println("Completed");
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                System.out.println("Error:" + throwable.getMessage());
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                System.out.println(integer);
            }
        });
    }


    private static Observable<Integer> createObserver() {
        return Observable.create(
                new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        if (!emitter.isDisposed()) {

                            for (int i = 0; i < 5; i++) {
                                int temp = new Random().nextInt(10);
                                if (temp > 8) {
//如果 value功，则创建一个异常
                                    emitter.onError(new Throwable("value >8"));
                                    break;
                                } else {
                                    emitter.onNext(temp);
//没有发生异常，正常结束
                                    if (i == 4) {
                                        emitter.onComplete();
                                    }
                                }

                            }
                        }
                    }
                }
        );
    }
}
